package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2Partition {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("partition")

    val sc = new SparkContext(conf)


    /**
     * 在资源充足的情况下，分区越多执行速度越快，（只需要保证每一个分区的数据量在128M是比较合适的）
     *
     * RDD由多个分区组成
     * 默认分区数
     * 1、如果使用spark读取hdfs中的文件默认一个block对应一个分区
     * 2、使用算子之后rdd分区的数量默认等于前一个rdd分区数
     *
     * 改变rdd分区数
     * 1、在读取文件时可以指定一个最小分区数（minPartitions）参数，最终的分区数时block数量的整数倍，比最小分区数大
     * 2、没有shuffle的算子不能改变分区数
     * 3、有shuffle的算子可以改变分区数
     */

    /*
     *指定一个目录读取数据 ，会自动读取目录下所有的文件
     */
    val linesRDD: RDD[String] = sc.textFile("data/words", 4)

    println(s"linesRDD分区数：${linesRDD.getNumPartitions}")

    val wordsRDD: RDD[String] = linesRDD.flatMap((line: String) => line.split(","))

    println(s"wordsRDD分区数：${wordsRDD.getNumPartitions}")

    /**
     * shuffle类的算子可以改变分区数，相当于mr中手动设置reduce的数量
     */
    val groupByRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy((word: String) => word, 100)

    println(s"groupByRDD分区数：${groupByRDD.getNumPartitions}")

    val countRDD: RDD[(String, Int)] = groupByRDD.map(kv => (kv._1, kv._2.size))

    println(s"countRDD分区数：${countRDD.getNumPartitions}")


    /**
     * 将数据保存到hdfs
     *
     */
    countRDD.saveAsTextFile("data/count")


    while (true) {}
  }

}
